package com.shujia.flink.window;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class Demo5WindowProcess {
    public static void main(String[] args) throws Exception {
        /*
         *实时统计班级的平均年龄，每隔5秒统计一次
         */

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> linesDS = env.socketTextStream("master", 8888);

        //解析数据
        DataStream<Tuple2<String, Integer>> clazzAndAgeDS = linesDS
                .map(line -> {
                    String[] split = line.split(",");
                    String clazz = split[4];
                    int age = Integer.parseInt(split[2]);
                    return Tuple2.of(clazz, age);
                }, Types.TUPLE(Types.STRING, Types.INT));

        //按照班级分组
        KeyedStream<Tuple2<String, Integer>, String> keyByDS = clazzAndAgeDS.keyBy(kv -> kv.f0);

        //划分窗口
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowDS = keyByDS
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)));

        DataStream<Tuple2<String, Double>> avgAgeDS = windowDS
                .process(new ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Double>, String, TimeWindow>() {
                    /**
                     * process方法每一个key每个窗口执行一次
                     * @param clazz 分组的key
                     * @param context 上下文对象。flink的环境
                     * @param datas  窗口内所有的数据
                     * @param out 同于将处理好的结果发送到下游
                     */
                    @Override
                    public void process(String clazz,
                                        ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Double>, String, TimeWindow>.Context context,
                                        Iterable<Tuple2<String, Integer>> datas,
                                        Collector<Tuple2<String, Double>> out) throws Exception {
                        //实时统计班级的平均年龄，每隔5秒统计一次

                        Double sumAge = 0.0;
                        int num = 0;
                        for (Tuple2<String, Integer> data : datas) {
                            Integer age = data.f1;
                            sumAge += age;
                            num++;
                        }
                        double avgAge = sumAge / num;

                        //将结果发送到下游
                        out.collect(Tuple2.of(clazz, avgAge));
                    }
                });

        avgAgeDS.print();


        env.execute();


    }
}
